package app.dwd;

import app.BaseSQLApp;
import common.Constant;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import utils.SQLUtil;

import java.time.Duration;

public class Dwd_03_DwdTradeOrderDetail extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_03_DwdTradeOrderDetail().init(
                3003,2,
                "Dwd_03_DwdTradeOrderDetail"
        );
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        tEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(5));
        //1、读取ods_db
        readOdsDb(tEnv,"Dwd_03_DwdTradeOrderDetail");

        //2、过滤订单明细
        Table orderDetail = tEnv.sqlQuery("select " +
                " data['id'] id,  " +
                " data['order_id'] order_id,  " +
                " data['course_id'] course_id,  " +
                " data['course_name'] course_name,  " +
                " data['user_id'] user_id,  " +
                " data['origin_amount'] origin_amount,  " +
                " data['coupon_reduce'] coupon_reduce,  " +
                " data['final_amount'] final_amount,  " +
                " data['session_id'] session_id,  " +
                " data['create_time'] create_time,  " +
                " data['update_time'] update_time,  " +
                " ts,  " +
                " pt  " +
                " from ods_db " +
                " where `database`='edu' " +
                " and `table`='order_detail' " +
                " and `type`='insert' ");
        tEnv.createTemporaryView("order_detail", orderDetail);

        // 3. 过滤订单表
        Table orderInfo = tEnv.sqlQuery("select " +
                " data['id'] id," +
                " data['order_status'] order_status," +
                " data['out_trade_no'] out_trade_no " +
                " data['trade_body'] trade_body " +
                " data['province_id'] province_id " +
                " from ods_db " +
                " where `database`='edu' " +
                " and `table`='order_info' " +
                " and `type`='insert' ");
        tEnv.createTemporaryView("order_info", orderInfo);
        // 4. 2 张表的 join
        Table result = tEnv.sqlQuery("select " +
                " od.id,  " +
                " od.course_id,  " +
                " od.course_name,  " +
                " od.order_id,  " +
                " od.user_id,  " +
                " od.origin_amount,  " +
                " od.coupon_reduce,  " +
                " od.final_amount,  " +
                " od.session_id,  " +
                " date_format(od.create_time, 'yyyy-MM-dd') date_id,  " +
                " od.create_time,  " +
                " oi.id" +
                " oi.order_status" +
                " oi.out_trade_no" +
                " oi.trade_body" +
                " oi.province_id" +
                " od.ts, " +
                " from order_detail od " +
                " join order_info oi on od.order_id=oi.id " );


        // 5. 写出到 kafka 中
        tEnv.executeSql("create table dwd_trade_order_detail(  " +
                " id string, " +
                " course_id string" +
                " course_name string" +
                " order_id string" +
                " user_id string" +
                " origin_amount string" +
                " coupon_reduce string" +
                " final_amount string" +
                " session_id string" +
                " create_time string" +
                " update_time string" +
                " order_status string" +
                " out_trade_no string" +
                " trade_body string" +
                " province_id string" +
                " ts bigint,  " +
                " primary key(id) not enforced  " +
                " )" + SQLUtil.getKafkaSink(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL));

        result.executeInsert("dwd_trade_order_detail");

    }
}
